package com.atguigu.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc DataStream实现WordCount（socket无界流）
 * @since 2024-11-23 10:21:49
 */
public class SocketUnboundedStream {

    public static void main(String[] args) throws Exception {
        // TODO 1. 创建执行环境
        // StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // IDEA运行时，也可以看到webui，一般用于本地测试
        // 需要引入一个依赖 flink-runtime-web
        // 在idea运行，不指定并行度，默认就是 电脑的 线程数，UI页面地址：http://localhost:8081/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(3);

        // TODO 2. 读取数据： socket
        // DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
        DataStreamSource<String> socketDS = env.socketTextStream("127.0.0.1", 7777);

        // TODO 3. 处理数据: 切换、转换、分组、聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                .flatMap(
                        new FlatMapFunction<String, Tuple2<String, Integer>>() {
                            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                                String[] words = value.split(" ");
                                for (String word : words) {
                                    out.collect(Tuple2.of(word, 1));
                                }
                            }
                        }
                )
                .setParallelism(2)
                .keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
                    public Object getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                })
                .sum(1)
                .setParallelism(3);

        // TODO 4. 输出
        sum.print();
        // TODO 5. 执行
        env.execute();
    }

}

/*
 * 并行度的优先级：
 * 代码：算子 > 代码(env) > 提交时指定 > 配置文件
 *
 * 说明：Flink程序的并行度可以通过多种方式指定，优先级从高到低分别是：
 * 1. 算子级别：可以通过算子的setParallelism()方法指定并行度
 * 2. env级别：可以通过env.setParallelism()方法指定并行度
 * 3. 提交时指定：可以通过命令行参数指定并行度
 * 4. 配置文件：可以通过flink-conf.yaml文件指定并行度
 */
